package org.raymond.iworks.study.mq.rabbitmq.stream.official;

import com.rabbitmq.stream.Environment;

import java.util.Scanner;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */

/**
 * 消费者在创建其订阅时提供初始数量的积分。信用表示允许代理发送给消费者的消息块。
 * 块是一批消息。这是RabbitMQ流中使用的存储和传输单元，即消息连续存储在一个块中，并作为块的一部分进行传递。根据入口的不同，一个块可以由一个到几千个消息组成。
 * 因此，如果消费者创建具有5个初始积分的订阅，则代理将发送5个消息块。经纪人每次交付一笔款项时都会减去一笔款项。当没有剩余的信用用于订阅时，代理停止发送消息。因此，在我们的示例中，代理将在交付5个块后停止为此订阅发送消息。这不是我们通常想要的，因此消费者可以向其订阅提供积分以获得更多消息。
 * 这取决于消费者(即客户端库和/或应用程序)提供积分，具体取决于它处理消息的速度。我们希望消息连续流动，所以一个好的经验法则是创建至少有2个积分的订阅，并为每个新的消息块提供一个积分。通过这样做，应该总是有一些消息在网络上流动，并且消费者应该一直很忙，而不是空闲。
 * 消费者可以选择代理如何使用此信用流机制向他们传递消息。这有助于避免超载或空闲的消费者。消费者信贷流如何向应用程序公开取决于客户端库，没有服务器端设置可以更改其行为。
 */
public class SuperStreamConsumer {
    public static void main(String[] args) {
        if (args.length != 1) {
            System.out.println("Specify the instance name!");
            System.exit(1);
        }
        String instanceName = args[0];
        try (Environment environment = Environment.builder().maxConsumersByConnection(1)
                .maxTrackingConsumersByConnection(1).build()) {

            // to exit properly when Ctrl-C-ed
            Runtime.getRuntime().addShutdownHook(new Thread(() -> environment.close()));

            String superStream = "invoices";
            String reference = "my-app";

            System.out.println("Starting consumer " + instanceName);
            environment.consumerBuilder().superStream(superStream).name(reference).singleActiveConsumer()
                    .autoTrackingStrategy().messageCountBeforeStorage(10).builder()
                    .messageHandler((context, message) -> {
                        System.out.println(
                                "Consumer " + instanceName + " received message " + message.getProperties()
                                        .getMessageId() + " from stream " + context.stream() + ".");
                    }).build();
            Scanner keyboard = new Scanner(System.in);
            keyboard.nextLine();
        }
    }
}
